package com.dhy.canal;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.aop.support.AopUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author 大忽悠
 * @create 2022/12/27 9:52
 */
@Slf4j
public class CanalMessageParser {

    /**
     * 解析拉取到的消息
     */
    public static Map<String, CanalMessageDispatcher.CanalEventHolder> parse(Message message){
        Map<String,CanalMessageDispatcher.CanalEventHolder> res=new HashMap<>();
        for (CanalEntry.Entry entry : message.getEntries()) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                //开启/关闭事务的实体类型，跳过
                continue;
            }
            //RowChange对象，包含了一行数据变化的所有特征
            //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                  log.error("解析canal事件时发生异常: ",e);
                  continue;
            }
            //获取操作类型：insert/update/delete类型
            CanalEntry.EventType eventType = rowChange.getEventType();
            //打印Header信息
            log.info("================》; binlog[{}:{}] , name[{},{}] , eventType : {}",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType);
            //判断是否是DDL语句
            if (rowChange.getIsDdl()) {
                System.out.println("================》;isDDL: true,sql:" + rowChange.getSql());
            }
            //解析DML事件
            doParseDML(eventType,entry.getHeader().getTableName(),rowChange,res);
        }
        return res;
    }

    private static void doParseDML(CanalEntry.EventType eventType, String tableName, CanalEntry.RowChange rowChange, Map<String, CanalMessageDispatcher.CanalEventHolder> res) {
        //获取RowChange对象里的每一行数据，打印出来
        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            CanalMessageDispatcher.CanalEventHolder canalEventHolder = CanalMessageDispatcher.CanalEventHolder.builder()
                    .eventType(eventType)
                    .beforeColumns(rowData.getBeforeColumnsList())
                    .afterColumns(rowData.getAfterColumnsList())
                    .build();
            res.put(tableName,canalEventHolder);
            logEvent(eventType,rowData);
        }
    }

    private static void logEvent(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        //如果是删除语句
        if (eventType == CanalEntry.EventType.DELETE) {
            printColumn(rowData.getBeforeColumnsList());
            //如果是新增语句
        } else if (eventType == CanalEntry.EventType.INSERT) {
            printColumn(rowData.getAfterColumnsList());
            //如果是更新的语句
        } else {
            //变更前的数据
            System.out.println("------->; before");
            printColumn(rowData.getBeforeColumnsList());
            //变更后的数据
            System.out.println("------->; after");
            printColumn(rowData.getAfterColumnsList());
        }
    }

    private static void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

    /**
     * 解析当前CanalMessageListener监听的表名
     */
    public static String parseListenTable(Object bean) {
        CanalTable canalTable = AopUtils.getTargetClass(bean).getAnnotation(CanalTable.class);
        String value = canalTable.value();
        if(StringUtils.isEmpty(value)){
            throw new IllegalArgumentException(bean.getClass().getName()+"类上的@CanalTable注解中监听的表名不能为空!");
        }
        //类名首字母小写作为表名
        return value;
    }
}
